package com.matrix.activemq.topic;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class ConsumerTool implements MessageListener, ExceptionListener{
    private String user = ActiveMQConnection.DEFAULT_USER;
    private String password = ActiveMQConnection.DEFAULT_PASSWORD;
    private String url = ActiveMQConnection.DEFAULT_BROKER_URL;
    private String subject = "TopicTest1";
    private Connection connection = null;
    private Destination destination = null;
    private Session session = null;
    private MessageConsumer consumer = null;
    public static Boolean isConnection = false;

    private void initialize() throws JMSException {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
        connection = connectionFactory.createConnection();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        destination = session.createTopic(subject);
        consumer = session.createConsumer(destination);
    }

    public void consumerMessage() throws JMSException {
        initialize();
        connection.start();
        consumer.setMessageListener((MessageListener) this);
        connection.setExceptionListener((ExceptionListener) this);
        isConnection = true;
        System.out.println("Consumer:->Begin Listening..." );
    }

    public void close() throws JMSException {
        System.out.println("Consumer:->Closing Connection!");
        if (consumer != null) {
            consumer.close();
        } else if (session != null) {
            session.close();
        } else if (connection != null) {
            connection.close();
        }
    }


    public void onMessage(Message message) {
        if (message instanceof TextMessage) {
            TextMessage txtmsg = (TextMessage) message;
            try {
                String text = txtmsg.getText();
                System.out.println("Consumer:->Received: " + text);
            } catch (JMSException e) {
                e.printStackTrace();
            }
            System.out.println();
        } else {
            System.out.println("Consumer:->Received: " + message);
        }
    }

    public void onException(JMSException e) {
        isConnection = false;
    }
}
